package com.gzc.just.play.last.war.common.thread;

import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 基于JDK21虚拟线程的消息处理器
 * 将消息处理从IO线程中分离，提高网络性能和响应能力
 */
public class VirtualThreadMessageProcessor {
    private static final Logger logger = LoggerFactory.getLogger(VirtualThreadMessageProcessor.class);
    
    // 消息处理队列
    private final BlockingQueue<MessageProcessingTask> messageQueue = new LinkedBlockingQueue<>();
    
    // 线程池管理
    private final ThreadFactory threadFactory;
    private final int threadPoolSize;
    private final Thread[] threads;
    
    // 线程计数器，用于命名
    private final AtomicInteger threadCounter = new AtomicInteger(0);
    
    // 处理器状态
    private volatile boolean running = false;
    
    /**
     * 构造函数
     * @param processorName 处理器名称
     * @param threadPoolSize 线程池大小
     */
    public VirtualThreadMessageProcessor(String processorName, int threadPoolSize) {
        this.threadPoolSize = threadPoolSize;
        this.threadFactory = r -> {
            Thread thread = Thread.ofVirtual().unstarted(r);
            thread.setName(processorName + "-VirtualThread-" + threadCounter.incrementAndGet());
            thread.setDaemon(true);
            return thread;
        };
        this.threads = new Thread[threadPoolSize];
    }
    
    /**
     * 启动消息处理器
     */
    public void start() {
        if (running) {
            logger.warn("Message processor is already running");
            return;
        }
        
        running = true;
        
        // 创建并启动所有虚拟线程
        for (int i = 0; i < threadPoolSize; i++) {
            threads[i] = threadFactory.newThread(this::processMessages);
            threads[i].start();
        }
        
        logger.info("Started {} virtual message processor threads", threadPoolSize);
    }
    
    /**
     * 停止消息处理器
     */
    public void stop() {
        if (!running) {
            return;
        }
        
        running = false;
        
        // 通知所有线程停止
        for (Thread thread : threads) {
            if (thread != null && thread.isAlive()) {
                thread.interrupt();
            }
        }
        
        // 等待所有线程结束
        for (Thread thread : threads) {
            try {
                if (thread != null) {
                    thread.join(5000); // 最多等待5秒
                }
            } catch (InterruptedException e) {
                logger.warn("Interrupted while waiting for thread to stop", e);
                Thread.currentThread().interrupt();
            }
        }
        
        logger.info("Stopped virtual message processor threads");
    }
    
    /**
     * 提交消息处理任务
     * @param messageId 消息ID
     * @param messageBytes 消息字节数组
     * @param context Netty通道上下文
     * @param handler 消息处理器
     */
    public void submitTask(int messageId, byte[] messageBytes, ChannelHandlerContext context, MessageHandler handler) {
        if (!running) {
            logger.warn("Message processor is not running, ignoring message ID: {}", messageId);
            return;
        }
        
        MessageProcessingTask task = new MessageProcessingTask(messageId, messageBytes, context, handler);
        
        try {
            messageQueue.put(task);
        } catch (InterruptedException e) {
            logger.error("Interrupted while submitting message processing task", e);
            Thread.currentThread().interrupt();
        }
    }
    
    /**
     * 获取当前队列大小
     * @return 队列中待处理的消息数量
     */
    public int getQueueSize() {
        return messageQueue.size();
    }
    
    /**
     * 处理消息的主循环
     * 在虚拟线程中执行
     */
    private void processMessages() {
        logger.info("Started virtual message processing thread: {}", Thread.currentThread().getName());
        
        try {
            while (running &&
                    !Thread.currentThread().isInterrupted()) {
                try {
                    // 从队列中取出消息处理任务
                    MessageProcessingTask task = messageQueue.take();
                    // 处理消息
                    task.run();
                } catch (InterruptedException e) {
                    logger.info("Message processing thread interrupted: {}", Thread.currentThread().getName());
                    break;
                } catch (Exception e) {
                    logger.error("Error processing message in virtual thread: {}", Thread.currentThread().getName(), e);
                }
            }
        } finally {
            logger.info("Stopped virtual message processing thread: {}", Thread.currentThread().getName());
        }
    }
}